
package cfar.distributed.reduce;

import cfar.actors.Location;
import cfar.actors.Message;
import cfar.actors.Node;
import cfar.actors.distributed.Signal;
import cfar.ff.data.Assocs.Assoc2;
import cfar.ff.data.List;
import cfar.ff.data.Maybe;
import cfar.ff.data.Stream;
import cfar.ff.fun.F0;
import cfar.ff.fun.F1;
import cfar.ff.fun.F2;
import cfar.ff.fun.F4;
import cfar.ff.fun.F5;
import cfar.ff.fun.Seq;
import cfar.ff.interfaces.Sequence;

import static cfar.ff.data.Maybe.*;
import static cfar.ff.fun.Curry.*;

import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.Iterator;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.logging.Level;
import java.util.logging.Logger;



/**
 *
 * @author badi
 */
public class RReduce<S1,S2,A> implements F5< Sequence<S1, Integer>
                                           , F1<List<A>, A>
                                           , F0<Sequence<S2,Node>>
                                           , F2<A,A,A>
                                           , Sequence<S1, A>
                                           , A
                                           >
{

    /**********************************************************************/
    public A f( Sequence<S1, Integer> redundancies
              , F1<List<A>, A>        chooser
              , F0<Sequence<S2, Node>>get_nodes
              , F2<A, A, A>           f
              , Sequence<S1, A>       seq         )
    {

        SentPackage p = gen_package(redundancies, chooser, get_nodes, f, seq);
        return p.call();
    }
    /**********************************************************************/


    private SentPackage gen_package( Sequence<S1, Integer> redundancies
                                           , F1<List<A>, A>        chooser
                                           , F0<Sequence<S2, Node>>get_nodes
                                           , F2<A, A, A>           f
                                           , Sequence<S1, A>       seq         )
    {
        Stream<SentPackage> packages = Stream.<SentPackage>nil();
        Sequence<S1,Assoc2<Integer,A>> zipped = Seq.zip(redundancies, seq);

        java.util.List<Location> locs = new java.util.LinkedList<Location>();
        for(Node n : get_nodes.call())
            locs.add(n.server_location());

        java.util.List<Assoc2<Integer,A>>
                zs =
                new java.util.LinkedList<Assoc2<Integer,A>>
                    ((Collection<? extends Assoc2<Integer, A>>) Seq.mkList(zipped).getList().toCollection());
        SentPackage p = new SentPackage(chooser, get_nodes, zs, f);

        return p;
    }


    private class SentPackage implements Serializable, F0<A> {

        final F1<List<A>, A> chooser;
        final F0<Sequence<S2,Node>> get_nodes;
        final java.util.List<Assoc2<Integer,A>> seq;
        final F2<A,A,A> reducer;

        public SentPackage( final F1<List<A>, A> chooser
                          , final F0<Sequence<S2,Node>> get_nodes
                          , final java.util.List<Assoc2<Integer,A>> seq
                          , final F2<A,A,A> reducer
                          )
        {
            this.chooser = chooser;
            this.get_nodes = get_nodes;
            this.seq = seq;
            this.reducer = reducer;
        }


        Maybe<Future<A>> calc_rest(SentPackage p, final Node node, ExecutorService pool){
            node.tell(new Message(Signal.WorkF0, p));
            F0<A> f = new F0<A>() {

                public A call() {
                    Assoc2<Maybe,A> a = (Assoc2<Maybe,A>) node.call().payload;
                    node.tell(Message.close); // cleanup
                    return a.$2;
                }
            };

            return just(pool.submit(f));
        }


        /* don't actually execute them yet. This allows the computations to
         * be built, the remaining ones sent out, then these ones evaluated */
        Assoc2<Integer,List<F0<A>>> duplicate_runs( Assoc2<Integer,A> first
                                                  , Assoc2<Integer,A> second
                                                  , Iterator<Node> nodes
                                                  , F2<A,A,A> reducer
                                                  )
        {
            int r = Math.max(first.$1, second.$1);
            List<F0<A>> runs = List.nil();
            for(int i = 0; i < r-1; i++){ // r-1 because we want to run one instance locally
                Node n = nodes.next();
                F0<A> run = mkRun(n, first.$2, second.$2, reducer);
                runs = runs.cons(run);
            }
            F0<A> local = mkLocalRun(first.$2, second.$2, reducer);
            runs = runs.cons(local);

            return new Assoc2<Integer, List<F0<A>>>(r, runs);
        }

        F0<A> mkLocalRun(final A a1, final A a2, final F2<A,A,A> f){
            return new F0<A>() {

                public A call() {
                    return f.f(a1, a2);
                }
            };
        }

        F0<A> mkRun(final Node n, final A a1, final A a2, final F2<A,A,A> f){
            return new F0<A>() {

                public A call() {
                    F0<A> f0 = curry(curry(f, a1), a2);
                    n.tell(new Message(Signal.WorkF0, f0));
                    A res = (A) n.call().payload;
                    n.tell(Message.close);
                    return res;
                }
            };
        }

        F1<F0<A>,A> eval = new F1<F0<A>, A>() {

            public A f(F0<A> f) {
                return f.call();
            }
        };

        public A call() {

            Assoc2<Integer, A> my_first;
            Assoc2<Integer, A> my_second;
            Maybe<Assoc2<Integer,A>> my_last = nothing();
            Maybe<Future<A>> rest = nothing();
            boolean recur = false;
            ExecutorService pool = Executors.newCachedThreadPool();


            if(seq.size() > 3){
                recur = true;

            // handle the case of an odd number of elements
            }else if (seq.size() == 3){
                my_last = just(seq.remove(2));
            }
            
            
            my_first = seq.remove(0);
            my_second = seq.remove(0);

            // get the nodes
            Iterator<Node> cycled_nodes = Seq.cycle(get_nodes.call());

            // get unevaluated function to solve the rest
            Assoc2<Integer,List<F0<A>>> runs = duplicate_runs(my_first, my_second, cycled_nodes, reducer);

            if(recur){
                //make the new package and send it off, calculating the remainder
                rest = calc_rest(new SentPackage(chooser, get_nodes, seq, reducer), cycled_nodes.next(), pool);
            }


            // now that I've recursed (if needed) I can run locally
            List<A> results = (List<A>) Seq.parMap(eval, runs.$2);
            A result = chooser.f(results);


            // handle the case that there was one left
            if(my_last.isJust){
                Assoc2<Integer,A> result2 = new Assoc2<Integer, A>(runs.$1, result);
                Assoc2<Integer,List<F0<A>>> last_runs = duplicate_runs(result2, my_last.fromJust(), cycled_nodes, reducer);
                List<A> rs = (List<A>) Seq.parMap(eval, last_runs.$2);
                result = chooser.f(rs);
            }

            // join with recured subproblem
            if(rest.isJust){
                try {
                    A next = rest.fromJust().get();
                    result = reducer.f(result, next);
                } catch (InterruptedException ex) {
                    Logger.getLogger(RReduce.class.getName()).log(Level.SEVERE, null, ex);
                } catch (ExecutionException ex) {
                    Logger.getLogger(RReduce.class.getName()).log(Level.SEVERE, null, ex);
                }
            }

            // ok, we're done!
            pool.shutdown();
            return result;
        }

    }
}
